package mq

import (
	log "ac-common-go/glog"
	"ac-julink/cache"
	"ac-julink/manager"
	"ac-julink/tool/util"
	"github.com/go-stomp/stomp"
	"github.com/go-stomp/stomp/frame"
	"net"
	"strings"
	"time"
)

const (
	RETRYSTEP = 3
)

type activeMqListener struct {
	listenerService MqListenerService
}

func (self *activeMqListener) Start(brokerUrl, destination, clientId string, context *cache.SyncContext, client *manager.AcRestClient, username, password string, mqSSL int) {
	defer func() {
		if pErr := recover(); pErr != nil {
			log.Errorf("active mq listener panic:%+v\n%s", pErr, util.PanicTrace())
		}
	}()
	log.Infof("activeMqListener start")
	if brokerUrl == "" || destination == "" {
		log.Error("mq listener start error, broker or destination is empty")
		return
	}
	brokerArr := strings.Split(brokerUrl, "*")
	if len(brokerArr) == 0 {
		log.Error("activemq.broker-url is error, can not start listener!!!")
		return
	}
	log.Infof("activeMQ username: %s; password: %s", username, password)
	var connOptions []func(*stomp.Conn) error = []func(*stomp.Conn) error{
		stomp.ConnOpt.Login(username, password),
		//stomp.ConnOpt.Login("admin", "admin"),
		stomp.ConnOpt.Host("/"),
		stomp.ConnOpt.Header("client-id", clientId),
	}
	var subOptions []func(*frame.Frame) error = []func(*frame.Frame) error{}
	errBrokerMap := make(map[string]int64)
	var sleepSeconds int64 = 0
	loop := 0
	var conn *stomp.Conn
	var err error
	for {
		brokerAddr := brokerArr[loop]
		log.Infof("activeMQ brokerAddr: %s", brokerAddr)
		retryTimes, existErr := errBrokerMap[brokerAddr]
		if existErr {
			sleepSeconds = RETRYSTEP * retryTimes
		}

		if sleepSeconds > 0 {
			log.Errorf("connect error before, will retry after %d seconds!!!", sleepSeconds)
			time.Sleep(time.Second * time.Duration(sleepSeconds))
		}
		log.Info("start a new mq connection")

		if mqSSL == 1 {
			conn, err = stomp.Dial("tcp", brokerAddr, connOptions...)
		} else {
			conn, err = Dial("tcp", brokerAddr, connOptions...)
		}

		if err != nil {
			errBrokerMap[brokerAddr] = retryTimes + 1
			log.Errorf("can not connect, err[%v], will try next!!!", err.Error())
			loop++
			if loop >= len(brokerArr) {
				loop = 0
			}
			continue
		} else {
			log.Infof("success to connect")
			delete(errBrokerMap, brokerAddr)
		}
		sub, err := conn.Subscribe(destination, stomp.AckAuto, subOptions...)
		if err != nil {
			log.Errorf("connect subscriber error")
			break
		}
		for {
			msg := <-sub.C
			if !sub.Active() {
				log.Errorf("subscribe is not active")
				break
			}
			self.listenerService.OnReceive(msg, context, client)
		}
	}
}
func Dial(network, addr string, opts ...func(*stomp.Conn) error) (*stomp.Conn, error) {
	c, err := net.Dial(network, addr)
	if err != nil {
		return nil, err
	}

	host, _, err := net.SplitHostPort(c.RemoteAddr().String())
	if err != nil {
		c.Close()
		return nil, err
	}

	// Add option to set host and make it the first option in list,
	// so that if host has been explicitly specified it will override.
	opts = append([](func(*stomp.Conn) error){stomp.ConnOpt.Host(host)}, opts...)

	return stomp.Connect(c, opts...)
}
